package com.schoolComments.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.schoolComments.dto.Result;
import com.schoolComments.entity.SeckillVoucher;
import com.schoolComments.entity.VoucherOrder;
import com.schoolComments.mapper.VoucherMapper;
import com.schoolComments.mapper.VoucherOrderMapper;
import com.schoolComments.service.IVoucherOrderService;
import com.schoolComments.utils.RedisIdWorker;
import com.schoolComments.utils.UserHolder;
import com.schoolComments.utils.lock.ILockImpl;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * <p>
 *  服务实现类
 * </p>
 */
@Service
@Slf4j
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {
    @Resource
    StringRedisTemplate stringRedisTemplate;
    @Resource
    private RedisIdWorker redisIdWorker;
    @Resource
    private VoucherMapper voucherMapper;
    @Resource VoucherOrderMapper voucherOrderMapper;
    @Resource
    private IVoucherOrderService voucherOrderService;
    @Resource
    private RedissonClient redissonClient;
    //异步处理的线程池
    private static final ExecutorService seckillVoucher_order_Executor =
            Executors.newSingleThreadExecutor();
    //lua脚本
    private static final DefaultRedisScript<Long> DEFAULT_REDIS_SCRIPT;
    //lua脚本在类初始化的时候就加载进来 提高性能(不用每次用的时候再去加载)
    static{
        //将lua脚本加载进来
        DEFAULT_REDIS_SCRIPT = new DefaultRedisScript<>();

        DEFAULT_REDIS_SCRIPT.setLocation(new ClassPathResource("seckillVoucher.lua"));
        DEFAULT_REDIS_SCRIPT.setResultType(Long.class);
    }
    //给线程池提交任务 应该在类初始化完成后就提交 因此使用@PostConstruct
    @PostConstruct
    private void init(){
        seckillVoucher_order_Executor.submit(new VoucherOrderHandler());
    }
    //类在初始化完成后 就应该执行VoucherOrderHandler中的run方法 既不断的从消息队列中取数据并处理
    private class VoucherOrderHandler implements Runnable{
        @Override
        public void run() {
            while(true){
                try {
                //从消息队列中取消息
                List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                        Consumer.from("g1", "c1"),
                        StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                        StreamOffset.create("stream.orders", ReadOffset.lastConsumed())
                );
                if (list == null || list.isEmpty()) {
                    //没取到
                    continue;
                }
                //取到数据
                //处理数据
                    MapRecord<String, Object, Object> entries = list.get(0);
                    Map<Object, Object> value = entries.getValue();
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
                    //创建订单
                    Integer count = voucherOrderMapper.insertOrder(voucherOrder);
                    //确认(ACK)
                    if (count == 1) {
                        stringRedisTemplate.opsForStream().acknowledge("stream.orders", "g1", entries.getId());
                    }
                } catch (Exception e) {
                    log.info("消息处理错误");
                    e.printStackTrace();
                    //处理异常消息
                    handlePendingList();
                }
            }
        }
    }

    private void handlePendingList() {
        while(true){
            try {
                //从pending-list中取消息
                List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                        Consumer.from("g1", "c1"),
                        StreamReadOptions.empty().count(1),
                        StreamOffset.create("stream.orders", ReadOffset.from("0"))
                );
                if (list == null || list.isEmpty()) {
                    //没取到 说明没错误要处理了(所有信息都确认完了) 结束就可以了
                    break;
                }
                //取到数据
                //处理数据
                MapRecord<String, Object, Object> entries = list.get(0);
                Map<Object, Object> value = entries.getValue();
                VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
                //创建订单
                Integer count = voucherOrderMapper.insertOrder(voucherOrder);
                //确认(ACK)
                if (count == 1) {
                    stringRedisTemplate.opsForStream().acknowledge("stream.orders", "g1", entries.getId());
                }
            } catch (Exception e) {
                log.info("消息处理错误");
            }
        }
    }


    @Override
    //版本一 加普通synchronize锁 无法实现分布式
    public Result addSeckillVoucherOrder(Long voucherId) {
        //根据id查询优惠券
        SeckillVoucher seckillVoucher = voucherMapper.querySeckillVocherById(voucherId);
        //秒杀是否开始
        if (!LocalDateTime.now().isAfter(seckillVoucher.getBeginTime())) {
            //没开始
            return Result.fail("没到开始时间");
        }
        //已经开始
        //是否还有券
        if (seckillVoucher.getStock() < 1) {
            //没有
            return Result.fail("被抢光了");
        }
        //有
        //这里实现一人一单功能
        Long id = UserHolder.getUser().getId();
        //这里的锁不要加在createOrder方法上(锁是this) 那样粒度太大了 就成了串行了
        //应该用一个锁的代码块 用常量池里的id作为锁 这样只会锁到相同的id  不同的用户之间是不干扰的 比较高效
        //intern()-->
        //这里要先提交事务 然后再释放锁 不然会出现并发问题 比如 两个张三线程在这 其中一个创建了订单 如果先释放锁 那可能事务还没提交 另一个张三线程就得到锁了 这个时候他去判断 数据库里是没数据的 所以他又可以下单了
        //因此 必须先提交事务 再释放锁
        synchronized (id.toString().intern()) {
            //这里createOrder()是有事务的 如果直接写return createOrder(seckillVoucher, voucherId); 就等于
            //this.createOrder() 那么这个事务就失效了 并没有加上事务 因为事务是spring用aop动态生成的
            //所以我们要得到代理对象 代理对象才是有事务的 这里获得代理对象有两种方法
            //1.直接获取代理对象 VoucherOrderServiceImpl voucherOrderService = (VoucherOrderServiceImpl) AopContext.currentProxy();
            //2.注入代理对象 直接使用 在这里相当于自己注入自己 (循环依赖 ..不会)
            return voucherOrderService.createOrder(seckillVoucher);
//            return createOrder(seckillVoucher, voucherId); 错误的 这样会使createOrder方法的事务失效
        }
    }

    //版本2 加分布式锁
    public Result addSeckillVoucherOrder2(Long voucherId) {
        //根据id查询优惠券
        SeckillVoucher seckillVoucher = voucherMapper.querySeckillVocherById(voucherId);
        //秒杀是否开始
        if (!LocalDateTime.now().isAfter(seckillVoucher.getBeginTime())) {
            //没开始
            return Result.fail("没到开始时间");
        }
        //已经开始
        //是否还有券
        if (seckillVoucher.getStock() < 1) {
            //没有
            return Result.fail("被抢光了");
        }
        //有
        //这里实现一人一单功能
        Long id = UserHolder.getUser().getId();
        //使用分布式锁实现一人一单
        ILockImpl iLock = new ILockImpl(stringRedisTemplate, id + "");
        boolean lock = iLock.getLock(30);
        if (!lock){
            //获取锁失败
            return Result.fail("不允许重复下单");
        }
        try {
            //获取锁成功 则创建订单
            return voucherOrderService.createOrder(seckillVoucher);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        finally {
            //释放锁
            iLock.unlock();
        }
    }

    //版本3 使用Redisson锁
    public Result addSeckillVoucherOrder3(Long voucherId) {
        //根据id查询优惠券
        SeckillVoucher seckillVoucher = voucherMapper.querySeckillVocherById(voucherId);
        //秒杀是否开始
        if (!LocalDateTime.now().isAfter(seckillVoucher.getBeginTime())) {
            //没开始
            return Result.fail("没到开始时间");
        }
        //已经开始
        //是否还有券
        if (seckillVoucher.getStock() < 1) {
            //没有
            return Result.fail("被抢光了");
        }
        //有
        //这里实现一人一单功能
        Long id = UserHolder.getUser().getId();
        //使用Redisson锁实现一人一单
        //使用RedissonClient创建锁对象
        RLock lock = redissonClient.getLock("lock:order:" + id);
        boolean success = lock.tryLock();
        if (!success){
            //获取锁失败
            return Result.fail("不允许重复下单");
        }
        try {
            //获取锁成功 则创建订单
            return voucherOrderService.createOrder(seckillVoucher);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        finally {
            //释放锁
            lock.unlock();
        }
    }

    @Transactional
    public Result createOrder(SeckillVoucher seckillVoucher){
        Long id = UserHolder.getUser().getId();
        Integer integer = voucherOrderMapper.queryOrderByIdAndVoucherId(id,seckillVoucher.getVoucherId());
        if (integer > 0) {
            //订单已经存在 不能再重复买了
            return Result.fail("不能重复购买");
        }
        //券的库存-1
        //@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@更新操作应该在数据库完成 如果在这做会有线程安全问题 比如有10个线程同时到达此处 现在有2张票 这样就会同时下10个订单 寄 所以这种更新操作还是放数据库里去吧
//        seckillVoucher.setStock(seckillVoucher.getStock() - 1);
        //用乐观锁解决票数超卖问题
        //————乐观锁是用在数据更新时候的 所以是写在sql语句中的
        Integer count = voucherMapper.subStock(seckillVoucher);
        if (count == 0) {
            //更新失败 说明券不足(多个线程抢一张券 只有一个线程能获得)
            return Result.fail("被抢光了");
        }
        //抢券成功的线程才能写入到订单中
        //向优惠券订单列表中写入数据
        VoucherOrder voucherOrder = new VoucherOrder();
        //准备一个全局唯一id 作为voucherOrder(优惠券订单)的主键
        Long voucherId1 = redisIdWorker.getId("voucherId");
        //当前抢购优惠券的用户的id
        Long userId = UserHolder.getUser().getId();
        voucherOrder.setId(voucherId1);
        voucherOrder.setUserId(userId);
        voucherOrder.setVoucherId(seckillVoucher.getVoucherId());
        //写入数据库
        voucherOrderMapper.insertOrder(voucherOrder);
        //返回订单id给前端
        return Result.ok(voucherId1);
    }

//    版本4 使用redis完成查询功能(lua脚本)  使用redis stream消息队列完成缓存功能
    public Result addSeckillVoucherOrder4(Long voucherId) {
        Long userId = UserHolder.getUser().getId();
        Long orderId = redisIdWorker.getId("voucherId");
        //执行lua脚本 判断是否有购买资格
        Long result = (Long) stringRedisTemplate.execute(
                DEFAULT_REDIS_SCRIPT,
                Collections.emptyList(),
                String.valueOf(voucherId), String.valueOf(userId), String.valueOf(orderId)
        );
        //没有购买资格
        if(result != null && result != 0L){
            return result == 1L?Result.fail("库存不足"):Result.fail("不能重复下单");
        }
        //有购买资格 说明在lua脚本中已经执行了相关购买操作(redis中减库存 加购买人信息) 并且已经将消息放入了消息队列
        //异步线程会处理消息队列中的消息 所以现在直接返回订单id给前端就行了
        //至于异步线程什么时候真的把数据保存到db 我们不关心
        return Result.ok(orderId);
    }
}
